package com.yingdong.gateserver.opcode;

import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.yingdong.gateserver.innter.handler.DisconnectedHandler;
import com.yingdong.gateserver.innter.handler.GateServerHandler;
import com.yingdong.gateserver.innter.handler.HeartBeatHandler;
import com.yingdong.gateserver.innter.message.GameMessage;
import com.yingdong.gateserver.net.GameSession;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;

public class Dispatcher {
	public static final int QUEUE_SIZE = 10000;
	static Log logger = LogFactory.getLog(Dispatcher.class);
	public static BlockingQueue<ByteBuf> receiveBuffers = new LinkedBlockingQueue<ByteBuf>(QUEUE_SIZE);
	public static BlockingQueue<ByteBuf> sendBuffers = new LinkedBlockingQueue<ByteBuf>(QUEUE_SIZE);
	public static Channel sceneServerSession;
	public static Map<Integer, GameSession> sessionMap = new ConcurrentHashMap<Integer, GameSession>();
	public static Map<Integer, Integer> reconnectedMap = new ConcurrentHashMap<Integer, Integer>();

	public static void init() {

		initHandlerTable();
		initThread();
	}

	public static GateServerHandler[] handlerTable = new GateServerHandler[3000];

	public static GateServerHandler[] handlerTable1 = new GateServerHandler[3000];

	private static void initHandlerTable() {
		handlerTable[InternalOpcode.HEARTBEAT] = new HeartBeatHandler();
		handlerTable1[InternalOpcode.DISCONNECT] = new DisconnectedHandler();
	}

	private static void initThread() {

		Thread readThread = new Thread(new Runnable() {

			@Override
			public void run() {
				while (true) {
					ByteBuf buffer = null;
					try {
						buffer = receiveBuffers.take();
						try {
							buffer.readShort();
							int length = buffer.readShort();
							int opcode = buffer.readInt();
							int sessionId = buffer.getInt(length);
							GameMessage message = new GameMessage(opcode, sessionId, buffer);
							if (opcode > 0 && Dispatcher.handlerTable[opcode] != null) {
								Dispatcher.handlerTable[opcode].handle(sessionId, message);
								buffer.release();
							} else {
								buffer.resetReaderIndex();
								Dispatcher.sceneServerSession.writeAndFlush(buffer);
							}
						} catch (Exception e1) {

							logger.debug("", e1);
						}
					} catch (Exception e) {

						logger.error("", e);
					}

				}

			}
		});
		readThread.start();
		Thread sendThread = new Thread(new Runnable() {

			@Override
			public void run() {
				while (true) {
					try {
						ByteBuf buffer = sendBuffers.take();
						int allength = buffer.readInt();
						int length = buffer.readInt();
						int opcode = buffer.readInt();
						buffer.readerIndex(length);
						// LoggerService.trace("opcode: " + opcode);
						while (buffer.isReadable()) {
							int session = buffer.readInt();
							GameSession gamesession = Dispatcher.sessionMap.get(session);
							if (gamesession != null) {
								io.netty.channel.Channel sessionChannel = Dispatcher.sessionMap.get(session).getChannel();
								ByteBuf buffer2 = buffer.slice(0, length);
								if (Dispatcher.handlerTable1[opcode] != null) {
									GameMessage message = new GameMessage(opcode, session, buffer2);
									Dispatcher.handlerTable1[opcode].handle(session, message);

								} else {
									sessionChannel.writeAndFlush(buffer2);
								}
							}
						}
					} catch (Exception e) {
						logger.error("", e);
					}

				}

			}
		});
		// sendThread.start();
		Thread checkThread = new Thread(new Runnable() {

			@Override
			public void run() {
				while (true) {
					try {
						// new CheckOnlineSessionsMessage(0,
						// sessionMap).sendToSceneServer();
						Thread.sleep(60000);
					} catch (InterruptedException e) {
						logger.error("", e);
					}
				}

			}

		});
		checkThread.start();
		// sendThread.start();

	}
}
